package com.atguigu.flink.chapter11.function;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @Author lzc
 * @Date 2022/7/13 13:57
 */
public class Flink02_Table {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("hello atguigu", 1L, 10),
            new WaterSensor("hello atguigu atguigu", 2L, 20),
            new WaterSensor("a b c", 3L, 30),
            new WaterSensor("aa cc dddd", 4L, 40)
        );
        
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromDataStream(stream);  // 把流转成一个动态表
        tEnv.createTemporaryView("sensor", table);
        
        
        // 1. 在tableApi中使用
        
        // 1.1 内联的方式使用
        /*table
            //            .joinLateral(call(MySplit.class, $("id")))  // 内连接
            .leftOuterJoinLateral(call(MySplit.class, $("id")))  // 内连接
            .select($("id"), $("word"), $("len"))
            .execute()
            .print();*/
        // 1.2 先注册后使用
        
        
        // 2. 在sql中使用
        tEnv.createTemporaryFunction("my_split", MySplit.class);
        
        // select .. form a join b on a.id=b.id 内连接
        // select ... from a,b where a.id=b.id  内连接
        /*tEnv.sqlQuery("select " +
                          " id, " +
                          " word, " +
                          " len " +
                          "from sensor " +
                          "join lateral table( my_split(id) ) on true")
            .execute()
            .print();*/
    
        /*tEnv.sqlQuery("select " +
                          " id, " +
                          " word, " +
                          " len " +
                          "from sensor " +
                          ", lateral table( my_split(id) )")
            .execute()
            .print();*/
    
        /*tEnv.sqlQuery("select " +
                          " id, " +
                          " word, " +
                          " len " +
                          "from sensor " +
                          "left join lateral table( my_split(id) ) on true")
            .execute()
            .print();*/
    
        tEnv.sqlQuery("select " +
                          " id, " +
                          " w, " +
                          " l " +
                          "from sensor " +
                          "left join lateral table( my_split(id) ) as T(w, l) on true")
            .execute()
            .print();
        
    }
    
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class MySplit extends TableFunction<Row> {
        public void eval(String s) {
            if (s != null) {
                if (s.contains("atguigu")) {
                    return;
                }
                String[] words = s.split(" ");
                for (String word : words) {
                    collect(Row.of(word, word.length()));
                }
            }
        }
    }
    
    
}
/*
"hello atguigu"
   hello   5
   atguigu 7
   
"hello atguigu"   hello   5
"hello atguigu"   atguigu 7

"hello atguigu atguigu"
    ....

 */